package com.lzg.okrx.observable

import com.lzy.okgo.exception.HttpException
import com.lzy.okgo.model.Response
import io.reactivex.Observable
import io.reactivex.Observer
import io.reactivex.disposables.Disposable
import io.reactivex.exceptions.CompositeException
import io.reactivex.exceptions.Exceptions
import io.reactivex.plugins.RxJavaPlugins

class BodyObservable<T>(
    private val upstream: Observable<Response<T>>
) : Observable<T>() {

    override fun subscribeActual(observer: Observer<in T>) {
        upstream.subscribe(BodyObserver(observer))
    }

    private class BodyObserver<R> internal constructor(
        private val observer: Observer<in R>
    ) : Observer<Response<R>> {

        private var terminated: Boolean = false

        override fun onSubscribe(disposable: Disposable) {
            observer.onSubscribe(disposable)
        }

        override fun onNext(response: Response<R>) {
            if (response.isSuccessful) {
                observer.onNext(response.body())
            } else {
                terminated = true
                val t = HttpException(response)
                try {
                    observer.onError(t)
                } catch (inner: Throwable) {
                    Exceptions.throwIfFatal(inner)
                    RxJavaPlugins.onError(CompositeException(t, inner))
                }

            }
        }

        override fun onComplete() {
            if (!terminated) {
                observer.onComplete()
            } else {
                // This should never happen! onNext handles and forwards errors automatically.
                val broken = AssertionError("This should never happen! Report as a bug with the full stacktrace.")
                RxJavaPlugins.onError(broken)
            }
        }

        override fun onError(throwable: Throwable) {
            if (!terminated) {
                observer.onError(throwable)
            } else {
                // This should never happen! onNext handles and forwards errors automatically.
                val broken = AssertionError("This should never happen! Report as a bug with the full stacktrace.")

                broken.initCause(throwable)
                RxJavaPlugins.onError(broken)
            }
        }
    }

}
